package drds.global_transaction.api.rpc.rpc_client;

import com.alibaba.fastjson.JSON;
import drds.common.Constants;
import drds.common.Json;
import drds.common.Log;
import drds.global_transaction.api.message_codec.RequestMessage;
import drds.global_transaction.api.message_codec.RequestMessages;
import drds.global_transaction.api.rpc.rpc_message.RpcInfo;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;

@Slf4j
public class SendRequestMessagesRunnable implements Runnable
{
	private RpcClient rpcClient;

	public SendRequestMessagesRunnable(RpcClient rpcClient)
	{
		this.rpcClient = rpcClient;
	}

	@Override
	public void run()
	{
		while (true)
		{
			synchronized (rpcClient.addressToRpcInfoQueueMapLock)
			{
				try
				{
					rpcClient.addressToRpcInfoQueueMapLock.wait(RpcClient.max_merge_send_mills);
				} catch (InterruptedException e)
				{
				}
			}
			// mark 'sending message'
			rpcClient.isSending = true;
			for (String address : rpcClient.addressToRpcInfoQueueMap.keySet())
			{
				BlockingQueue<RpcInfo> rpcInfoQueue = rpcClient.addressToRpcInfoQueueMap.get(address);
				if (rpcInfoQueue.isEmpty())
				{
					continue;
				}
				RequestMessages requestMessages = new RequestMessages();
				while (!rpcInfoQueue.isEmpty())
				{
					RpcInfo rpcInfo = rpcInfoQueue.poll();
					requestMessages.requestMessageList.add((RequestMessage) rpcInfo.getMessage());
					requestMessages.requestMessageIdList.add(rpcInfo.getId());
				}
				Channel channel = rpcClient.connect(Nets.toInetSocketAddress(address));// 原来是address
				try
				{
					if (Constants.developMode)
					{
						log.info("待发送数据:" + requestMessages.getClass().getSimpleName());
						Log.output(log, Json.formate(JSON.toJSONString(requestMessages)));
					}
					rpcClient.synchronousSendRequest(channel, requestMessages);
				} catch (FrameworkException e)
				{
					if (e.getFrameworkErrorCode() == FrameworkErrorCode.ChannelIsNotWritable && address != null)
					{
						rpcClient.returnChannel(address, channel);
					}
					log.error("", "client merge call failed", e);
				}
			}
			rpcClient.isSending = false;
		}
	}
}
